package app.dwm;

import app.dwd.BaseAppFollowV2;
import bean.OrderDetail;
import bean.OrderInfo;
import bean.OrderWide;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import common.Constant;
import function.DIMAsyncFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;
import util.DimUtil;
import util.GmallSinkUtil;
import util.JDBCUtil;
import util.RedisUtil;

import java.sql.Connection;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import static common.Constant.*;

public class DWMOrderWideAppFollow_Cache_Async extends BaseAppFollowV2 {
    //1.order_info和order_detail进行interval join;需要从kafka获取两个流;baseapp不符合要求,重构
    public static void main(String[] args) {
        new DWMOrderWideAppFollow_Cache_Async().init(5555,
                                                     1,
                                                     "DWMOrderWideAppFollow_Cache_Async",
                                                     "DWMOrderWideAppFollow_Cache_Async",
                                                     TOPIC_DWD_ORDER_INFO,
                                                     TOPIC_DWD_ORDER_DETAIL
        );
    }

    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streamKafkaSourcehashMap) {
        DataStreamSource<String> dwdOrderInfoStream = streamKafkaSourcehashMap.get(TOPIC_DWD_ORDER_INFO);
        DataStreamSource<String> dwdOrderDetailStream = streamKafkaSourcehashMap.get(TOPIC_DWD_ORDER_DETAIL);

        //intervaljoin前提:keyby,水印
        SingleOutputStreamOperator<OrderWide> orderWideStream = twoStreamIntervalJoin(dwdOrderInfoStream, dwdOrderDetailStream);

        SingleOutputStreamOperator<OrderWide> orderWideSingleOutputStreamOperator = dimJoin(orderWideStream);

        GmallSinkUtil.sendToKafka(orderWideSingleOutputStreamOperator.map(s -> JSON.toJSONString(s)), Constant.TOPIC_DWM_ORDER_WIDE);
    }

    private SingleOutputStreamOperator<OrderWide> dimJoin(SingleOutputStreamOperator<OrderWide> orderWideStream) {
        //使用异步方式进行(查询)join
        return AsyncDataStream.unorderedWait(orderWideStream,//哪个流进行异步
                                             new DIMAsyncFunction<OrderWide>() {
                                                 @Override
                                                 public void addDim(Connection conn, Jedis jedis, OrderWide orderWide, ResultFuture<OrderWide> resultFuture) throws Exception {
                                                     //读取userinfo
                                                     JSONObject userInfo = DimUtil.readDim(conn, jedis, DIM_USER_INFO, orderWide.getUser_id(), 1);
                                                     orderWide.setUser_gender(userInfo.getString("GENDER"));
                                                     orderWide.calcUser_Age(userInfo.getString("BIRTHDAY"));

                                                     // 2. 读取省份的信息
                                                     JSONObject provinceInfo = DimUtil.readDim(conn, jedis, Constant.DIM_BASE_PROVINCE, orderWide.getProvince_id(), 1);
                                                     orderWide.setProvince_name(provinceInfo.getString("NAME"));
                                                     orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
                                                     orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));

                                                     orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));

                                                     // 3. 读取sku信息
                                                     JSONObject skuInfo = DimUtil.readDim(conn, jedis, Constant.DIM_SKU_INFO, orderWide.getSku_id(), 1);
                                                     orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                                                     orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                                                     orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                                                     orderWide.setTm_id(skuInfo.getLong("TM_ID"));
                                                     orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                                                     // 4. 读取tm信息
                                                     JSONObject tmInfo = DimUtil.readDim(conn, jedis, Constant.DIM_BASE_TRADEMARK, orderWide.getTm_id(), 1);
                                                     orderWide.setTm_name(tmInfo.getString("TM_NAME"));

                                                     // 5. 读取spu信息
                                                     JSONObject spuInfo = DimUtil.readDim(conn, jedis, Constant.DIM_SPU_INFO, orderWide.getSpu_id(), 1);
                                                     orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                                                     // 6. 读取c3信息
                                                     JSONObject c3Info = DimUtil.readDim(conn, jedis, Constant.DIM_BASE_CATEGORY3, orderWide.getCategory3_id(), 1);
                                                     orderWide.setCategory3_name(c3Info.getString("NAME"));

                                                     // 把补齐维度数据的OrderWide交给resultFuture, 会自动后序的流中
                                                     resultFuture.complete(Collections.singleton(orderWide));

                                                 }
                                             }, //自定义异步处理数据的函数
                                             120,//异步io请求超时的时间
                                             TimeUnit.SECONDS);

    }

    private SingleOutputStreamOperator<OrderWide> twoStreamIntervalJoin(DataStreamSource<String> dwdOrderInfoStream, DataStreamSource<String> dwdOrderDetailStream) {
        KeyedStream<OrderInfo, Long> orderInfoLongKeyedStream = dwdOrderInfoStream
            .map(value -> JSON.parseObject(value, OrderInfo.class))
            .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> obj.getCreate_ts()))
            .keyBy(OrderInfo::getId);

        KeyedStream<OrderDetail, Long> orderDetailLongKeyedStream = dwdOrderDetailStream
            .map(value -> JSON.parseObject(value, OrderDetail.class))
            .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> obj.getCreate_ts()))
            .keyBy(OrderDetail::getOrder_id);

        return orderInfoLongKeyedStream
            .intervalJoin(orderDetailLongKeyedStream)
            .between(Time.minutes(-5), Time.minutes(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });


    }
}
